package net.gd.utils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.DeleteMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import net.gd.domain.UdfModelLog;
import net.sf.json.JSONObject;

/**
 * 
 * @author cxq
 */
public class ZeppelinApiUtils {

  private static Logger logger = Logger.getLogger(ZeppelinApiUtils.class);

  public static final String SCALA = "scala";
  public static final String R = "r";
  public static final String PYTHON = "python";

  public static final String PREFIX_SCALA = "%spark";
  public static final String PREFIX_R = "%spark.r";
  public static final String PREFIX_PYTHON = "%spark.pyspark";

  static String serverPrefix = ConfigurationManager.getProperty("zeppelin.server");

  // 创建note, post
  static final String API_CREATE_NOTE = "/api/notebook";

  // 创建脚本内容, post
  static final String API_CREATE_PARAGRAPH = "/api/notebook/[noteId]/paragraph";

  // 同步运行脚本, post
  static final String API_SYNC_RUN_PARAGRAPH = "/api/notebook/run/[noteId]/[paragraphId]";

  // 删除note, delete
  static final String API_DELETE_NOTE = "/api/notebook/[noteId]";


  public static final String CODE = "code";
  public static final String MSG = "msg";
  public static final String DATA = "data";

  public static final String SUCCESS = "SUCCESS";
  public static final String ERROR = "ERROR";

  /**
   * @param noteName note name
   * @param paragraphs 脚本
   * @return paragraph id
   */
  public static String createParagraphs(String noteId, String paragraphs) {
    Map<String, String> paras = new HashMap<String, String>();
    paras.put("title", "abd-paragraphs");
    paras.put("text", paragraphs);
    String paragraphJson = JSONObject.fromObject(paras).toString();
    logger.info(paragraphJson);
    String url = serverPrefix + API_CREATE_PARAGRAPH.replace("[noteId]", noteId);
    try {
      Map<String, Object> rs = doPost(url, paragraphJson);
      return rs.get("body").toString();
    } catch (Exception e) {
      logger.error("create noteboot failed.", e);
      return null;
    }
  }

  /**
   * @param noteName note name
   * @param paragraphs 脚本
   * @return note id
   */
  public static String createNote() {
    Map<String, Object> note = new HashMap<String, Object>();
    note.put("name", "abd-" + System.currentTimeMillis());
    String noteJson = JSONObject.fromObject(note).toString();
    logger.info(noteJson);
    try {
      Map<String, Object> rs = doPost(serverPrefix + API_CREATE_NOTE, noteJson);
      return rs.get("body").toString();
    } catch (Exception e) {
      logger.error("create noteboot failed.", e);
      return null;
    }
  }

  public static void deleteNote(String noteId) {
    String url = serverPrefix + API_DELETE_NOTE.replace("[noteId]", noteId);
    try {
      doDelete(url);
    } catch (Exception e) {
      logger.warn("delete failed", e);
    }
  }

  public static Map<String, Object> runParagraph(String noteId, String paragraphId) {
    String url = serverPrefix
        + API_SYNC_RUN_PARAGRAPH.replace("[noteId]", noteId).replace("[paragraphId]", paragraphId);
    try {
      return doPost(url, null);
    } catch (Exception e) {
      logger.error("run failed.", e);
    }
    return null;

  }

  /**
   * 运行模型并返回结果
   * @param langType
   * @param content
   * @return
   */
  public static UdfModelLog runModelContent(String langType, String content) {
    UdfModelLog log = new UdfModelLog();
    log.setLogId("log_" + new SimpleDateFormat("yyyyMMddHHmmssSSS").format(new Date()));
    log.setOperationDatetime(new Date());
    log.setOperationType(langType);
    log.setLangType(langType);
    log.setModelContent(content);
    
    String paragraph = null;
    if (SCALA.equals(langType)) {
      paragraph = PREFIX_SCALA + "\n" + content;
    } else if (PYTHON.equals(langType)) {
      paragraph = PREFIX_PYTHON + "\n" + content;
    } else if (R.equals(langType)) {
      paragraph = PREFIX_R + "\n" + content;
    } else {
      log.setOperationStatus(ERROR);
      log.setOperationResult("unsupport language " + langType);
      logger.warn("unsupport language " + langType);
      return log;
    }
    if(StringUtils.isBlank(content)) {
      log.setOperationStatus(ERROR);
      log.setOperationResult("model content is empty");
      return log;
    }
    
    String noteId = createNote();
    String paragraphId = createParagraphs(noteId, paragraph);
    logger.info("noteId: " + noteId + " paragraphId: " + paragraphId);
    Map<String, Object> runRs = runParagraph(noteId, paragraphId);
    deleteNote(noteId);
    Map body = (Map) runRs.get("body");
    String code = body.get("code").toString();
    List<Map> msgList = (List<Map>) body.get("msg");
    String codeInfo = msgList.get(0).get("data").toString();
    log.setOperationStatus(code);
    log.setOperationResult(codeInfo);
    return log;
  }
  

  public static void main(String[] args) {
//    UdfModelLog log =
//        runModelContent(SCALA, "val curr = System.currentTimeMillis();\nprintln(curr);");
    String noteId = createNote();
    System.out.println(noteId);
//    System.out.println(JSONObject.fromObject(log).toString());
  }

  public static Map<String, Object> doDelete(String url) throws Exception {
    DeleteMethod method = new DeleteMethod(url);
    try {
      HttpClient client = new HttpClient();
      // 设置为默认的重试策略
      method.getParams().setParameter(HttpMethodParams.RETRY_HANDLER,
          new DefaultHttpMethodRetryHandler());
      int rspCode = client.executeMethod(method);
      logger.info("post: " + url + " status: " + rspCode);
      String resp = method.getResponseBodyAsString();
      logger.info("resp: " + resp);
      if (resp != null) {
        return JSONObject.fromObject(resp);
      }
      return null;
    } catch (Exception e) {
      throw e;
    } finally {
      method.releaseConnection();
    }

  }

  public static Map<String, Object> doPost(String url, String jsonInput) throws Exception {
    PostMethod method = new PostMethod(url);
    try {
      HttpClient client = new HttpClient();
      method.setRequestHeader("Accept", "application/json; charset=UTF-8");
      // 设置为默认的重试策略
      method.getParams().setParameter(HttpMethodParams.RETRY_HANDLER,
          new DefaultHttpMethodRetryHandler(0, false));
      if (jsonInput != null) {
        method.setRequestEntity(new StringRequestEntity(jsonInput, "application/json", "utf-8"));
      }
      int rspCode = client.executeMethod(method);
      logger.info("post: " + url + " status: " + rspCode);
      String resp = method.getResponseBodyAsString();
      logger.info("resp: " + resp);
      if (resp != null) {
        return JSONObject.fromObject(resp);
      }
      return null;
    } catch (Exception e) {
      throw e;
    } finally {
      method.releaseConnection();
    }

  }

}
